package com.yiyuan.app.mqtt;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.mqtt.server.callback.MessageListener;
import com.alibaba.mqtt.server.model.MessageProperties;
import com.guaike.core.redis.annotation.TryLock;
import com.guaike.core.redis.enums.LockModel;
import com.yiyuan.app.dao.GConfDeviceDAO;
import com.yiyuan.app.dao.GConfRoomDeviceDAO;
import com.yiyuan.app.dao.GRecordGameDAO;
import com.yiyuan.app.dao.GRecordMqttMessageDAO;
import com.yiyuan.app.entity.constant.MQTTCommandCodeConstant;
import com.yiyuan.app.entity.constant.MQTTTopicConstant;
import com.yiyuan.app.entity.enums.MessageTypeEnum;
import com.yiyuan.app.entity.po.GConfDevicePO;
import com.yiyuan.app.entity.po.GConfRoomDevicePO;
import com.yiyuan.app.entity.po.GRecordGamePO;
import com.yiyuan.app.entity.po.GRecordMqttMessagePO;
import com.yiyuan.app.service.MQTTService;
import lombok.AllArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.*;

@Slf4j
@Component
@AllArgsConstructor
public class DeviceListener implements MessageListener {
    private static final Set<String> commandCodeSet = new HashSet<>(Arrays.asList(MQTTCommandCodeConstant.PUSH_COINS_CODE, MQTTCommandCodeConstant.PUSH_COINS_RESULT_CODE, MQTTCommandCodeConstant.MACHINE_STATUS_CODE, MQTTCommandCodeConstant.PUSH_COINS_PRIZE_CODE));

    private final MQTTService mqttService;
    private final GRecordMqttMessageDAO gRecordMqttMessageDAO;
    private final GRecordGameDAO gRecordGameDAO;
    private final GConfDeviceDAO gConfDeviceDAO;
    private final GConfRoomDeviceDAO gConfRoomDeviceDAO;
    private final MQTTProducer messageProducer;
    private final ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 100, 1L, TimeUnit.MINUTES, new LinkedBlockingQueue<>(), Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());


    @SneakyThrows
    @Override
    @TryLock(lockKey = "#msgId", lockModel = LockModel.REENTRANT)
    public void process(String msgId, MessageProperties messageProperties, byte[] payload) {
        executor.execute(() -> {
            try {
                String msg = new String(payload, StandardCharsets.UTF_8);
                String[] split = messageProperties.getSecondTopic().split(MQTTTopicConstant.SEPARATOR_KEY);
                String commandCode = split[3];
                if (!commandCodeSet.contains(commandCode)) {
                    return;
                }
                log.info("Receive message : msgId = {}, messageProperties = {}, payload = {}", msgId, JSONObject.toJSONString(messageProperties), msg);
                GRecordMqttMessagePO messagePO = new GRecordMqttMessagePO();
                messagePO.setMessageId(msgId);
                messagePO.setParentTopic(messageProperties.getFirstTopic());
                messagePO.setChildTopic(messageProperties.getSecondTopic());
                messagePO.setMessage(msg);
                messagePO.setStatus(Boolean.FALSE);
                messagePO.setCommandCode(commandCode);
                String clientId;
                String deviceNumber;
//            String ackTopic;
//            String message;
                if (split[1].equals("state")) {
                    clientId = split[2];
                    if (commandCode.equals(MQTTCommandCodeConstant.PUSH_COINS_PRIZE_CODE)) {
                        msg = msg.replace("{", "");
                        msg = msg.replace("}", "");
                        String[] split1 = msg.split(";");
                        String[] split2 = split1[1].split("=");
                        if (!"6".equals(split2[1])) {
                            return;
                        }
                        messagePO.setMessageType(MessageTypeEnum.game.getCode());
                    } else if (commandCode.equals(MQTTCommandCodeConstant.MACHINE_STATUS_CODE)) {
                        String[] strings = clientId.split("@@@");
                        String serialNumber = strings[1];
                        GConfDevicePO devicePO = gConfDeviceDAO.findOne(GConfDevicePO::getSerialNumber, serialNumber);
                        if (Objects.isNull(devicePO)) {
                            return;
                        }
                        if (msg.equals("0") && Objects.isNull(devicePO.getRepairTime())) {
                            return;
                        }
                        if (!msg.equals("0") && !Objects.isNull(devicePO.getRepairTime())) {
                            return;
                        }
                        messagePO.setMessageType(MessageTypeEnum.device_status.getCode());
                    }
//                ackTopic = messageProperties.getFirstTopic() + MQTTTopicConstant.SEPARATOR_KEY + clientId + "/sub/" + messagePO.getCommandCode() + "ack";
//                message = messagePO.getCommandCode();
                } else {
                    messagePO.setMessageType(MessageTypeEnum.game.getCode());
                    clientId = split[1];
                    //                    message = messagePO.getMessage();
                    //                    ackTopic = messageProperties.getFirstTopic() + MQTTTopicConstant.SEPARATOR_KEY + clientId + "/pub/" + messagePO.getCommandCode();
                    //                    message = messagePO.getCommandCode();
                    //                    ackTopic = messageProperties.getFirstTopic() + MQTTTopicConstant.SEPARATOR_KEY + clientId + "/pub/" + messagePO.getCommandCode() + "ack";
                }
                String[] strings = clientId.split("@@@");
                deviceNumber = strings[1];
                if (messagePO.getMessageType().equals(MessageTypeEnum.game.getCode())) {
                    GConfDevicePO devicePO = gConfDeviceDAO.findOne(GConfDevicePO::getSerialNumber, deviceNumber);
                    GConfRoomDevicePO roomDevicePO = gConfRoomDeviceDAO.findOne(GConfRoomDevicePO::getDeviceId, devicePO.getId());
                    GRecordGamePO gamePO = gRecordGameDAO.findPushCoinsLatestGaming(roomDevicePO.getRoomId(), roomDevicePO.getPositionNo());
                    if (Objects.nonNull(gamePO)) {
                        messagePO.setGameId(gamePO.getId());
                    } else {
                        GRecordGamePO recordGamePO = gRecordGameDAO.findRoomLatestHistory(roomDevicePO.getRoomId(), roomDevicePO.getPositionNo());
                        messagePO.setGameId(recordGamePO.getId());
                    }
                }
                messagePO.setSerialNumber(deviceNumber);
                gRecordMqttMessageDAO.save(messagePO);
                if (commandCode.equals(MQTTCommandCodeConstant.PUSH_COINS_CODE)) {
                    mqttService.dealPushCoins(messagePO);
                } else if (commandCode.equals(MQTTCommandCodeConstant.PUSH_COINS_RESULT_CODE)) {
                    mqttService.dealPushCoinsResult(messagePO);
                } else if (commandCode.equals(MQTTCommandCodeConstant.PUSH_COINS_PRIZE_CODE)) {
                    mqttService.dealPrizeResult(messagePO);
                } else if (commandCode.equals(MQTTCommandCodeConstant.MACHINE_STATUS_CODE)) {
                    mqttService.dealMachineStatus(messagePO);
                }
            } catch (Exception e) {
                if (e.getMessage().contains("Duplicate entry")) {
                    return;
                }
                log.error("DeviceListener error={}", e);
            }
        });
    }
}
